package net.bwie.realtime.jtp.dwd.trade.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 识别是否首次评论：
 *   - is_new_comment = 1 首次评论
 *   - is_new_comment = 0 非首次评论 或者其他事件
 */

public class IsNewCommentProcessFunction extends KeyedProcessFunction<String,String,String> {

    //step1-声明状态
    private transient ValueState<String> isNewCommentState;

    @Override
    public void open(Configuration parameters) throws Exception {
        //step2-初始化状态
        isNewCommentState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("is_new_comment", String.class)
        );
    }

    @Override
    public void processElement(String s, KeyedProcessFunction<String, String, String>.Context context, Collector<String> collector) throws Exception {
        //step3-解析字段值
        JSONObject jsonObject = JSON.parseObject(s);

        //提取用户id和主播id,把用户id和主播id作为唯一标识符
        String user_id = jsonObject.getString("user_id");
        String anchor_id = jsonObject.getString("anchor_id");
        String live_session_id = jsonObject.getString("live_session_id");
        //step4-获取状态值--判断是不是评论事件
        String event_type = jsonObject.getString("event_type");
        //是评论事件的话我们进行处理，不是不做操作
        if(event_type != null && event_type.equals("评论")){
            //把用户id和主播id作为唯一标识符
            String key = user_id + "_" + anchor_id + "_" + live_session_id;

            //给isOrderState打标记，如果为null，说明这是用户的第一次下单
            if(isNewCommentState.value()==null){
                //首次评论
                jsonObject.put("is_new_comment",1);
                //同时把状态更新成1
                isNewCommentState.update(key);
            }else {
                //非首次评论
                jsonObject.put("is_new_comment",0);
            }
        }else{
            //不是评论，不做操作
            jsonObject.put("is_new_comment",0);
        }

        collector.collect(jsonObject.toJSONString());
    }
}
